Node / meshtastic / Meshtastic-Android / files / core / data / src / commonMain / kotlin / org / meshtastic / core / data / manager / MqttManagerImpl.kt

Displaying Raw • Download

T8b949e/*

* Copyright (c) 2025-2026 Meshtastic LLC

*

* This program is free software: you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation, either version 3 of the License, or

* (at your option) any later version.

*

* This program is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

*

* You should have received a copy of the GNU General Public License

* along with this program. If not, see <https://www.gnu.org/licenses/>.

*/

Tff7b72package T7ee787org.meshtastic.core.data.manager

Tff7b72import T7ee787co.touchlab.kermit.Logger

Tff7b72import T7ee787co.touchlab.kermit.Severity

Tff7b72import T7ee787kotlinx.coroutines.CoroutineScope

Tff7b72import T7ee787kotlinx.coroutines.Job

Tff7b72import T7ee787kotlinx.coroutines.flow.MutableStateFlow

Tff7b72import T7ee787kotlinx.coroutines.flow.SharingStarted

Tff7b72import T7ee787kotlinx.coroutines.flow.StateFlow

Tff7b72import T7ee787kotlinx.coroutines.flow.catch

Tff7b72import T7ee787kotlinx.coroutines.flow.combine

Tff7b72import T7ee787kotlinx.coroutines.flow.launchIn

Tff7b72import T7ee787kotlinx.coroutines.flow.onEach

Tff7b72import T7ee787kotlinx.coroutines.flow.stateIn

Tff7b72import T7ee787org.koin.core.annotation.Named

Tff7b72import T7ee787org.koin.core.annotation.Single

Tff7b72import T7ee787org.meshtastic.core.model.MqttConnectionState

Tff7b72import T7ee787org.meshtastic.core.network.repository.MQTTRepository

Tff7b72import T7ee787org.meshtastic.core.repository.MqttManager

Tff7b72import T7ee787org.meshtastic.core.repository.PacketHandler

Tff7b72import T7ee787org.meshtastic.core.repository.ServiceRepository

Tff7b72import T7ee787org.meshtastic.mqtt.ConnectionState

Tff7b72import T7ee787org.meshtastic.mqtt.MqttException

Tff7b72import T7ee787org.meshtastic.proto.MqttClientProxyMessage

Tff7b72import T7ee787org.meshtastic.proto.ToRadio

Tf0883e@Single

Tff7b72class T56d364MqttManagerImplTb4b4b4(

Tff7b72private Tff7b72val Te6edf3mqttRepositoryTb4b4b4: Te6edf3MQTTRepositoryTb4b4b4,

Tff7b72private Tff7b72val Te6edf3packetHandlerTb4b4b4: Te6edf3PacketHandlerTb4b4b4,

Tff7b72private Tff7b72val Te6edf3serviceRepositoryTb4b4b4: Te6edf3ServiceRepositoryTb4b4b4,

Tf0883e@NamedTb4b4b4(Ta5d6ff"Ta5d6ffServiceScopeTa5d6ff"Tb4b4b4) Tff7b72private Tff7b72val Te6edf3scopeTb4b4b4: Te6edf3CoroutineScopeTb4b4b4,

Tb4b4b4) Tb4b4b4: Te6edf3MqttManager Tb4b4b4{

Tff7b72private Tff7b72var Te6edf3mqttMessageFlowTb4b4b4: Te6edf3Job? Tff7b72= Tff7b72null

Tff7b72private Tff7b72val Te6edf3proxyActive Tff7b72= Te6edf3MutableStateFlowTb4b4b4(Tff7b72falseTb4b4b4)

Tff7b72override Tff7b72val Te6edf3mqttConnectionStateTb4b4b4: Te6edf3StateFlowTff7b72<Te6edf3MqttConnectionStateTff7b72> Tff7b72=

Te6edf3combineTb4b4b4(Te6edf3proxyActiveTb4b4b4, Te6edf3mqttRepositoryTb4b4b4.Te6edf3connectionStateTb4b4b4) Tb4b4b4{ Te6edf3activeTb4b4b4, Te6edf3libState Tff7b72-Tff7b72>

Tff7b72if Tb4b4b4(Tff7b72!Te6edf3activeTb4b4b4) Te6edf3MqttConnectionStateTb4b4b4.Te6edf3INACTIVE Tff7b72else Te6edf3libStateTb4b4b4.Te6edf3toAppStateTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tb4b4b4.Te6edf3stateInTb4b4b4(Te6edf3scopeTb4b4b4, Te6edf3SharingStartedTb4b4b4.Te6edf3EagerlyTb4b4b4, Te6edf3MqttConnectionStateTb4b4b4.Te6edf3INACTIVETb4b4b4)

Tff7b72override Tff7b72fun Td2a8ffstartProxyTb4b4b4(Te6edf3enabledTb4b4b4: Tffa657BooleanTb4b4b4, Te6edf3proxyToClientEnabledTb4b4b4: Tffa657BooleanTb4b4b4) Tb4b4b4{

Tff7b72if Tb4b4b4(Te6edf3mqttMessageFlowTff7b72?.Te6edf3isActive Tff7b72=Tff7b72= Tff7b72trueTb4b4b4) Tff7b72return

Tff7b72if Tb4b4b4(Te6edf3enabled Tff7b72&Tff7b72& Te6edf3proxyToClientEnabledTb4b4b4) Tb4b4b4{

Te6edf3proxyActiveTb4b4b4.Te6edf3value Tff7b72= Tff7b72true

Te6edf3mqttMessageFlow Tff7b72=

Te6edf3mqttRepositoryTb4b4b4.Te6edf3proxyMessageFlow

Tb4b4b4.Te6edf3onEach Tb4b4b4{ Te6edf3message Tff7b72-Tff7b72> Te6edf3packetHandlerTb4b4b4.Te6edf3sendToRadioTb4b4b4(Te6edf3ToRadioTb4b4b4(Te6edf3mqttClientProxyMessage Tff7b72= Te6edf3messageTb4b4b4)Tb4b4b4) Tb4b4b4}

Tb4b4b4.Te6edf3catch Tb4b4b4{ Te6edf3throwable Tff7b72-Tff7b72>

Te6edf3proxyActiveTb4b4b4.Te6edf3value Tff7b72= Tff7b72false

Tff7b72val Te6edf3message Tff7b72=

Tff7b72when Tb4b4b4(Te6edf3throwableTb4b4b4) Tb4b4b4{

Tff7b72is Te6edf3MqttExceptionTb4b4b4.Te6edf3ConnectionRejected Tff7b72-Tff7b72> Ta5d6ff"Ta5d6ffMQTT: connection rejected (check credentials)Ta5d6ff"

Tff7b72is Te6edf3MqttExceptionTb4b4b4.Te6edf3ConnectionLost Tff7b72-Tff7b72> Ta5d6ff"Ta5d6ffMQTT: connection lostTa5d6ff"

Tff7b72else Tff7b72-Tff7b72> Ta5d6ff"Ta5d6ffMQTT proxy failed: Tffd700${Te6edf3throwableTb4b4b4.Te6edf3messageTffd700}Ta5d6ff"

Tb4b4b4}

Te6edf3serviceRepositoryTb4b4b4.Te6edf3setErrorMessageTb4b4b4(Te6edf3text Tff7b72= Te6edf3messageTb4b4b4, Te6edf3severity Tff7b72= Te6edf3SeverityTb4b4b4.Te6edf3WarnTb4b4b4)

Tb4b4b4}

Tb4b4b4.Te6edf3launchInTb4b4b4(Te6edf3scopeTb4b4b4)

Tb4b4b4}

Tb4b4b4}

Tff7b72override Tff7b72fun Td2a8ffstopTb4b4b4(Tb4b4b4) Tb4b4b4{

Tff7b72if Tb4b4b4(Te6edf3mqttMessageFlowTff7b72?.Te6edf3isActive Tff7b72=Tff7b72= Tff7b72trueTb4b4b4) Tb4b4b4{

Te6edf3LoggerTb4b4b4.Te6edf3i Tb4b4b4{ Ta5d6ff"Ta5d6ffStopping MqttClientProxyTa5d6ff" Tb4b4b4}

Te6edf3mqttMessageFlowTff7b72?.Te6edf3cancelTb4b4b4(Tb4b4b4)

Te6edf3mqttMessageFlow Tff7b72= Tff7b72null

Tb4b4b4}

Te6edf3proxyActiveTb4b4b4.Te6edf3value Tff7b72= Tff7b72false

Tb4b4b4}

Tff7b72override Tff7b72fun Td2a8ffhandleMqttProxyMessageTb4b4b4(Te6edf3messageTb4b4b4: Te6edf3MqttClientProxyMessageTb4b4b4) Tb4b4b4{

Tff7b72val Te6edf3topic Tff7b72= Te6edf3messageTb4b4b4.Te6edf3topic

Te6edf3LoggerTb4b4b4.Te6edf3d Tb4b4b4{ Ta5d6ff"Ta5d6ff[mqttClientProxyMessage] Tffd700$Te6edf3topicTa5d6ff" Tb4b4b4}

Tff7b72val Te6edf3retained Tff7b72= Te6edf3messageTb4b4b4.Te6edf3retained Tff7b72=Tff7b72= Tff7b72true

Tff7b72when Tb4b4b4{

Te6edf3messageTb4b4b4.Te6edf3text Tff7b72!Tff7b72= Tff7b72null Tff7b72-Tff7b72> Tb4b4b4{

Te6edf3mqttRepositoryTb4b4b4.Te6edf3publishTb4b4b4(Te6edf3topicTb4b4b4, Te6edf3messageTb4b4b4.Te6edf3textTff7b72!!Tb4b4b4.Te6edf3encodeToByteArrayTb4b4b4(Tb4b4b4)Tb4b4b4, Te6edf3retainedTb4b4b4)

Tb4b4b4}

Te6edf3messageTb4b4b4.Te6edf3data_ Tff7b72!Tff7b72= Tff7b72null Tff7b72-Tff7b72> Tb4b4b4{

Te6edf3mqttRepositoryTb4b4b4.Te6edf3publishTb4b4b4(Te6edf3topicTb4b4b4, Te6edf3messageTb4b4b4.Te6edf3data_Tff7b72!!Tb4b4b4.Te6edf3toByteArrayTb4b4b4(Tb4b4b4)Tb4b4b4, Te6edf3retainedTb4b4b4)

Tb4b4b4}

Tff7b72else Tff7b72-Tff7b72> Tb4b4b4{Tb4b4b4}

Tb4b4b4}

Tb4b4b4}

Tff7b72private Tff7b72fun Te6edf3ConnectionStateTb4b4b4.Td2a8fftoAppStateTb4b4b4(Tb4b4b4)Tb4b4b4: Te6edf3MqttConnectionState Tff7b72= Tff7b72when Tb4b4b4(Tff7b72thisTb4b4b4) Tb4b4b4{

Te6edf3ConnectionStateTb4b4b4.Te6edf3DISCONNECTED Tff7b72-Tff7b72> Te6edf3MqttConnectionStateTb4b4b4.Te6edf3DISCONNECTED

Te6edf3ConnectionStateTb4b4b4.Te6edf3CONNECTING Tff7b72-Tff7b72> Te6edf3MqttConnectionStateTb4b4b4.Te6edf3CONNECTING

Te6edf3ConnectionStateTb4b4b4.Te6edf3CONNECTED Tff7b72-Tff7b72> Te6edf3MqttConnectionStateTb4b4b4.Te6edf3CONNECTED

Te6edf3ConnectionStateTb4b4b4.Te6edf3RECONNECTING Tff7b72-Tff7b72> Te6edf3MqttConnectionStateTb4b4b4.Te6edf3RECONNECTING

Tb4b4b4}

Tb4b4b4}